package student.convert

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaProducer}
import util.RandDomData

import java.util.Properties

object TestSinkKafka {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val properties = new Properties()
    properties.setProperty("bootstrap.servers","node1:9092")

    env.addSource(new RandDomData)
      .map(data=>data.toString)
      .addSink(new FlinkKafkaProducer[String]("test",new SimpleStringSchema(),properties))

    env.addSource(new RandDomData)
      .map(data=>data.toString)
      .print()

    env.execute()
  }
}
